package com.study;

import java.util.concurrent.*;
import java.util.concurrent.Flow.*;

public class BackpressureExample {
    public static void main(String[] args) throws InterruptedException {
        // Create a custom publisher
        CustomPublisher<Integer> publisher = new CustomPublisher<>();

        // Create a subscriber and register it with the publisher
        Subscriber<Integer> subscriber = new Subscriber<>() {
            private Subscription subscription;
            private final ExecutorService executorService = Executors.newFixedThreadPool(4);

            @Override
            public void onSubscribe(Subscription subscription) {
                this.subscription = subscription;
                subscription.request(1);
            }

            @Override
            public void onNext(Integer item) {
                System.out.println("Received: " + item);
                executorService.submit(() -> {
                    try {
                        Thread.sleep(1000); // Simulate slow processing
                        System.out.println("Processed: " + item);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    subscription.request(1);
                });
            }

            @Override
            public void onError(Throwable throwable) {
                System.err.println("Error: " + throwable.getMessage());
                executorService.shutdown();
            }

            @Override
            public void onComplete() {
                System.out.println("Completed");
                executorService.shutdown();
            }
        };

        publisher.subscribe(subscriber);

        // Publish items
        for (int i = 1; i <= 10; i++) {
            publisher.publish(i);
        }

        // Wait for subscriber to finish processing and close the publisher
        Thread.sleep(15000);
        publisher.close();
    }
}

class CustomPublisher<T> implements Publisher<T> {
    private final SubmissionPublisher<T> submissionPublisher;

    public CustomPublisher() {
        this.submissionPublisher = new SubmissionPublisher<>();
    }

    @Override
    public void subscribe(Subscriber<? super T> subscriber) {
        submissionPublisher.subscribe(subscriber);
    }

    public void publish(T item) {
        submissionPublisher.submit(item);
    }

    public void close() {
        submissionPublisher.close();
    }
}

